package org.iscas.tcse.datasample;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import redis.clients.jedis.Jedis;
import redis.clients.jedis.Transaction;

import java.util.List;

public class SampleController extends Thread {

    private static final Logger LOG = LoggerFactory.getLogger(SampleController.class);

    private final double lowerBound;
    private final String redisHost;
    private final int redisPort;

    private final double alpha;

    private final String sourceSpeedKey;
    private final String mlSpeedKey;

    private final String sampleRatioKey;

    private final int syncInterval;

    private volatile boolean isRunning = true;

    SampleController(SamplerConf conf) {
        this.lowerBound = conf.lowerBound;
        this.alpha = conf.alpha;
        this.redisHost = conf.redisHost;
        this.redisPort = conf.redisPort;
        this.sourceSpeedKey = conf.sourceSpeedKey;
        this.mlSpeedKey = conf.mlSpeedKey;
        this.sampleRatioKey = conf.sampleRatioKey;
        this.syncInterval = conf.syncInterval;
    }

    @Override
    public void run() {
        Jedis jedis = new Jedis(redisHost, redisPort);
        while (isRunning) {
            Transaction tx = jedis.multi();
            tx.get(sourceSpeedKey);
            tx.get(mlSpeedKey);
            List<Object> result = tx.exec();
            if (result == null || result.size() != 2) {
                continue;
            }

            double sourceRate = Double.parseDouble((String) result.get(0));
            double mlRate = Double.parseDouble((String) result.get(1));
            double sampleRatio = Math.max(lowerBound, alpha * mlRate / sourceRate);
            tx.close();
            jedis.set(sampleRatioKey, String.valueOf(sampleRatio));
            try {
                Thread.sleep(syncInterval);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        jedis.close();
    }

    public void stopRunning() {
        isRunning = false;
    }
}
